package baseFx.common.mq;

import baseFx.common.AutoCloseable;
import baseFx.common.*;
import baseFx.common.concurrent.Task;
import baseFx.common.context.BeanLife;
import baseFx.common.context.TypeResolver;
import baseFx.common.utils.ClassUtils;
import baseFx.common.utils.StringUtils;

import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Logger;

public class DefaultMqBeanDispatcher implements MqBeanDispatcher, AutoCloseable {
    private static Logger logger = Logger.getLogger(DefaultMqBeanDispatcher.class.getTypeName());
    private static String errkey = UUID.randomUUID().toString();
    private static String beforeKey = UUID.randomUUID().toString();
    private static String afterKey = UUID.randomUUID().toString();
    private static boolean isShutdown;
    private TypeResolver resolver;
    private Map<String, ClassMqBean> suberToCmb = new HashMap<>(8);
    private Map<String, List<ClassMqBean>> beanToCmb = new HashMap<>(8);
    private List<MqSubscriber> subscribers = new ArrayList<>();
    private AtomicInteger regCount = new AtomicInteger();

    public EventDispatcher<MqBeanBefore> evBefore = new EventDispatcher<>(beforeKey);
    public EventDispatcher<MqBeanAfter> evAfter = new EventDispatcher<>(afterKey);
    public EventDispatcher<MqBeanError> evError = new EventDispatcher<>(errkey);

    static {
        Runtime.getRuntime().addShutdownHook(new Thread(() -> isShutdown = true));
    }

    public DefaultMqBeanDispatcher(TypeResolver resolver) {
        this.resolver = resolver;
    }

    @Override
    public void push(MqBean bean) {
        String queue = getQueue(bean.getClass());
        try (MqPublisher mq = getPublisher(bean.getClass())) {
            mq.post(queue, valueSerialize(bean));
        }
    }

    @Override
    public void register(Class<?> clazz) {
        if (check(clazz) == false) {
            return;
        }
        List<Method> methods = new ArrayList<>();
        Class<?>[] interfacelist = clazz.getInterfaces();
        for (Class<?> intf : interfacelist) {
            if (MqBeanSubscriber.class.isAssignableFrom(intf) == false) {
                continue;
            }
            methods.addAll(Stream.create(intf.getDeclaredMethods()).toList());
        }
        if (methods.isEmpty()) {
            return;
        }
        methods = Stream.create(methods).filter(i -> {
            if (i.getParameterCount() != 1) {
                return false;
            }
            return MqBean.class.isAssignableFrom(i.getParameters()[0].getType());
        }).toList();
        methods = Stream.create(methods).map(i -> {
            try {
                return clazz.getDeclaredMethod(i.getName(), i.getParameterTypes()[0]);
            } catch (NoSuchMethodException e) {
                return null;
            }
        }).ifNotNull().distinct(i -> StringUtils.join("$", i.getName(), i.getParameterTypes()[0].getTypeName().hashCode())).toList();

        ClassMqBean cmb = new ClassMqBean();
        cmb.clazz = clazz;
        cmb.name = getName(clazz);
        for (Method method : methods) {
            Class<?> beanClass = method.getParameterTypes()[0];
            if (cmb.methodMap.containsKey(beanClass.getTypeName()) == false) {
                cmb.methodMap.put(beanClass.getTypeName(), new ArrayList<>());
                cmb.mbClassList.add(beanClass);
            }
            cmb.methodMap.get(beanClass.getTypeName()).add(method);
        }
        suberToCmb.put(cmb.clazz.getTypeName(), cmb);
        resolver.register(def -> {
            def.beanType = cmb.clazz;
            def.life = BeanLife.context;
        });
    }

    public void register(String pkgName) {
        register(pkgName, null);
    }

    public void register(String pkgName, Action1T<List<Class<?>>> done) {
        regCount.getAndAdd(1);
        Var<List<Class<?>>> list = new Var<>(new ArrayList<>());
        Task.run(() -> {
            List<Class<?>> rs = ClassUtils.getClassByPackage(pkgName, true);
            if (rs != null) {
                list.set(rs);
                for (Class<?> item : list.get()) {
                    register(item);
                }
            }
        }).onCompleted(t -> {
            if (t.getError() != null) {
                t.getError().printStackTrace();
            }
            if (done != null) {
                done.execute(list.get());
            }
            regCount.getAndDecrement();
        });
    }

    public void start() {
        Task.run(() -> {
            while (regCount.get() != 0) {
                Task.delay(100);
            }
            for (ClassMqBean cmb : suberToCmb.values()) {
                start(cmb.clazz);
            }
        }).onCompleted(t -> {
            if (t.getError() != null) {
                logger.severe(t.getError().getMessage());
            }
        }).await();
    }

    public void start(Class<?> clazz) {
        ClassMqBean cmb = suberToCmb.get(clazz.getTypeName());
        if (cmb.isStarted) {
            return;
        }
        for (Class<?> beanClass : cmb.mbClassList) {
            String queue = getQueue(beanClass);
            if (beanToCmb.containsKey(queue) == false) {
                beanToCmb.put(queue, new ArrayList<>());
            }
            beanToCmb.get(queue).add(cmb);
            MqSubscriber suber = getSubscriber(cmb.clazz, beanClass);
            if (suber == null) {
                continue;
            }
            if (subscribers.contains(suber) == false) {
                suber.eventReceived().add(this::onMsg);
                suber.eventError().add(this::onError);
                suber.declare(queue);
                subscribers.add(suber);
            }
            int count = getConsumeCount(cmb.clazz, beanClass);
            for (int i = 0; i < count; i++) {
                suber.consume(queue, cmb.name, cmb);
            }
        }
        cmb.isStarted = true;
    }

    private void onError(Object sender, SubscriberError e) {
        ClassMqBean cmb = (ClassMqBean) e.state;
        MqBeanError args = new MqBeanError(cmb.clazz, e.error);
        evError.fire(errkey, this, args);
        e.isRetry = args.isRetry;
    }

    private void onMsg(Object sender, MqMessage message) {
        ClassMqBean cmb = (ClassMqBean) message.state;
        Class<?> beanClass = Stream.create(cmb.mbClassList).first(i -> message.queue.equalsIgnoreCase(getQueue(i))).orElse(null);
        if (beanClass == null) {
            return;
        }
        try (TypeResolver sub = resolver.sub()) {
            MqBean data = (MqBean) valueDeserialize(message.data);
            MqBeanSubscriber suber = (MqBeanSubscriber) sub.resolve(cmb.clazz);
            MqBeanBefore before = new MqBeanBefore(cmb.clazz, suber, data);
            List<Method> methods = cmb.methodMap.get(beanClass.getTypeName());
            evBefore.fire(beforeKey, this, before);
            if (before.cancel) {
                return;
            }
            try {
                for (Method method : methods) {
                    method.invoke(suber, data);
                }
            } catch (Exception e) {
                message.reconsume = suber.onError(data, e);
            }
            if (message.reconsume == false) {
                MqBeanAfter after = new MqBeanAfter(cmb.clazz, suber, data);
                evAfter.fire(afterKey, this, after);
            }
        }
    }

    private String valueSerialize(Object value) {
        Map<String, String> msg = new HashMap<>(2);
        msg.put("data", value instanceof String ? value.toString() : serialize(value));
        msg.put("type", value.getClass().getTypeName());
        return serialize(msg);
    }

    private Object valueDeserialize(String value) {
        Map map = (Map) deserialize(value, Map.class);
        Class<?> clazz = ClassUtils.getClassForName(map.get("type").toString());
        if (clazz == String.class) {
            return value;
        }
        return deserialize(map.get("data").toString(), clazz);
    }

    protected String serialize(Object value) {
        return ClassUtils.serialize(value);
    }

    protected Object deserialize(String value, Class<?> clazz) {
        return ClassUtils.deserialize(value);
    }

    protected int getConsumeCount(Class<?> suberClass, Class<?> beanClass) {
        return 1;
    }

    protected String getName(Class<?> suberClass) {
        int hash = suberClass.getTypeName().hashCode();
        String code = String.valueOf(hash).replace("-", "");
        if (hash < 0) {
            code = "X" + code;
        }
        return suberClass.getSimpleName() + "." + code;
    }

    protected String getQueue(Class<?> beanClass) {
        MqBeanSetting attr = beanClass.getAnnotation(MqBeanSetting.class);
        if (attr != null) {
            return attr.queue();
        }
        int hash = beanClass.getTypeName().hashCode();
        String code = String.valueOf(hash).replace("-", "");
        if (hash < 0) {
            code = "X" + code;
        }
        return beanClass.getSimpleName() + "." + code;
    }

    protected MqSubscriber getSubscriber(Class<?> suberClass, Class<?> beanClass) {
        return resolver.resolve(MqSubscriber.class);
    }

    protected MqPublisher getPublisher(Class<?> beanClass) {
        return resolver.resolve(MqPublisher.class);
    }

    protected boolean check(Class<?> clazz) {
        if (suberToCmb.containsKey(clazz.getTypeName())) {
            return false;
        }
        if (Modifier.isAbstract(clazz.getModifiers())) {
            return false;
        }
        if (Modifier.isInterface(clazz.getModifiers())) {
            return false;
        }
        if (Modifier.isPrivate(clazz.getModifiers())) {
            return false;
        }
        if (MqBeanSubscriber.class.isAssignableFrom(clazz) == false) {
            return false;
        }
        if (clazz.getTypeName().contains("$") && Modifier.isStatic(clazz.getModifiers()) == false) {
            return false;
        }
        return true;
    }

    @Override
    public void close() {
        for (MqSubscriber item : subscribers) {
            item.close();
        }
        suberToCmb.clear();
        beanToCmb.clear();
        resolver = null;
    }

    private static class ClassMqBean {
        public boolean isStarted;
        public String name;
        public Class<?> clazz;
        public Map<String, List<Method>> methodMap = new HashMap<>();
        public List<Class<?>> mbClassList = new ArrayList<>();
    }
}
